package com.yisearch.search.utils;

import org.frameworkset.elasticsearch.ElasticSearchHelper;
import org.frameworkset.elasticsearch.client.DB2ESImportBuilder;
import org.frameworkset.elasticsearch.client.DataStream;
import org.frameworkset.elasticsearch.client.ExportResultHandler;
import org.frameworkset.elasticsearch.client.TaskCommand;
import org.frameworkset.elasticsearch.client.schedule.ImportIncreamentConfig;
import org.frameworkset.runtime.CommonLauncher;

/*
 * 数据灌入es工具类
 * @author yp
 * @date 2019/10/29
 * @param
 * @return
 */
public class ImportDataUtil {

    /**
     * 数据全量导入
     * @param dropIndice
     */
    public static void fullImportData(boolean dropIndice) {
        DB2ESImportBuilder importBuilder = DB2ESImportBuilder.newInstance();
        dropIndice = CommonLauncher.getBooleanAttribute("dropIndice", false);//同时指定了默认值
        int batchSize = CommonLauncher.getIntProperty("batchSize", 1);//同时指定了默认值
        int queueSize = CommonLauncher.getIntProperty("queueSize", 50);//同时指定了默认值
        int workThreads = CommonLauncher.getIntProperty("workThreads", 20);//同时指定了默认值
        //增量定时任务不要删表，但是可以通过删表来做初始化操作
        if (dropIndice) {
            try {
                //清除测试表,导入的时候回重建表，测试的时候加上为了看测试效果，实际线上环境不要删表
                String repsonse = ElasticSearchHelper.getRestClientUtil().dropIndice("product");
                System.out.println(repsonse);
            } catch (Exception e) {
            }
        }
        //设置外部数据源
        importBuilder.setDbName("yisearch")
                .setDbDriver("com.mysql.cj.jdbc.Driver") //数据库驱动程序，必须导入相关数据库的驱动jar包
                .setDbUrl("jdbc:mysql://0.0.0.0:3306/yisearch?useCursorFetch=true")//通过useCursorFetch=true启用mysql的游标fetch机制，否则会有严重的性能隐患，useCursorFetch必须和jdbcFetchSize参数配合使用，否则不会生效
                .setDbUser("root")
                .setDbPassword("123456")
                .setValidateSQL("select 1")
                .setUsePool(false);//是否使用连接池

        //设置数据bulk导入任务结果处理回调函数，对每次bulk任务的结果进行成功和失败反馈，然后针对失败的bulk任务通过error方法进行相应处理
        importBuilder.setExportResultHandler(new ExportResultHandler<String, String>() {
            @Override
            public void success(TaskCommand<String, String> taskCommand, String result) {
                String datas = taskCommand.getDatas();//执行的批量数据
//				System.out.println(result);//打印成功结果
            }

            @Override
            public void error(TaskCommand<String, String> taskCommand, String result) {
                //任务执行完毕，但是结果中包含错误信息
                //具体怎么处理失败数据可以自行决定,下面的示例显示重新导入失败数据的逻辑：
                // 从result中分析出导入失败的记录，然后重新构建data，设置到taskCommand中，重新导入，
                // 支持的导入次数由getMaxRetry方法返回的数字决定
                // String failDatas = ...;
                //taskCommand.setDatas(failDatas);
                //taskCommand.execute();
                String datas = taskCommand.getDatas();//执行的批量数据
//				System.out.println(result);//打印成功结果
            }

            @Override
            public void exception(TaskCommand<String, String> taskCommand, Exception exception) {
                //任务执行抛出异常，失败处理方法
                exception.printStackTrace();
            }

            /**
             * 如果对于执行有错误的任务，可以进行修正后重新执行，通过本方法
             * 返回允许的最大重试次数
             * @return
             */
            @Override
            public int getMaxRetry() {
                return -1;
            }
        });

        //指定导入数据的sql语句，必填项，可以设置自己的提取逻辑，
        // 设置增量变量log_id，增量变量名称#[log_id]可以多次出现在sql语句的不同位置中，例如：
        // select * from td_sm_log where log_id > #[log_id] and parent_id = #[log_id]
        // log_id和数据库对应的字段一致,就不需要设置setNumberLastValueColumn和setNumberLastValueColumn信息，
        // 但是需要设置setLastValueType告诉工具增量字段的类型

        importBuilder.setSql("select * from product ");

//		importBuilder.setSql("select * from td_sm_log ");
        /**
         * es相关配置
         */
        importBuilder
                .setIndex("product") //必填项
                .setIndexType("normal") //必填项
				.setRefreshOption("refresh")//可选项，null表示不实时刷新，importBuilder.setRefreshOption("refresh");表示实时刷新
                .setUseJavaName(false)//可选项,将数据库字段名称转换为java驼峰规范的名称，true转换，false不转换，默认false，例如:doc_id -> docId
                .setUseLowcase(false)  //可选项，true 列名称转小写，false列名称不转换小写，默认false，只要在UseJavaName为false的情况下，配置才起作用
                .setPrintTaskLog(true) //可选项，true 打印任务执行日志（耗时，处理记录数） false 不打印，默认值false
                .setBatchSize(batchSize);  //可选项,批量导入es的记录数，默认为-1，逐条处理，> 0时批量处理


        //定时任务配置结束
//
//		//设置任务执行拦截器，可以添加多个
//		importBuilder.addCallInterceptor(new CallInterceptor() {
//			@Override
//			public void preCall(TaskContext taskContext) {
//				System.out.println("preCall");
//			}
//
//			@Override
//			public void afterCall(TaskContext taskContext) {
//				System.out.println("afterCall");
//			}
//
//			@Override
//			public void throwException(TaskContext taskContext, Exception e) {
//				System.out.println("throwException");
//			}
//		}).addCallInterceptor(new CallInterceptor() {
//			@Override
//			public void preCall(TaskContext taskContext) {
//				System.out.println("preCall 1");
//			}
//
//			@Override
//			public void afterCall(TaskContext taskContext) {
//				System.out.println("afterCall 1");
//			}
//
//			@Override
//			public void throwException(TaskContext taskContext, Exception e) {
//				System.out.println("throwException 1");
//			}
//		});
//		//设置任务执行拦截器结束，可以添加多个
        //增量配置开始
//		importBuilder.setNumberLastValueColumn("log_id");//手动指定数字增量查询字段，默认采用上面设置的sql语句中的增量变量名称作为增量查询字段的名称，指定以后就用指定的字段
//		importBuilder.setDateLastValueColumn("log_id");//手动指定日期增量查询字段，默认采用上面设置的sql语句中的增量变量名称作为增量查询字段的名称，指定以后就用指定的字段
        importBuilder.setFromFirst(true);//任务重启时，重新开始采集数据，true 重新开始，false不重新开始，适合于每次全量导入数据的情况，如果是全量导入，可以先删除原来的索引数据
        importBuilder.setLastValueStorePath("logtable_import");//记录上次采集的增量字段值的文件路径，作为下次增量（或者重启后）采集数据的起点，不同的任务这个路径要不一样
//		importBuilder.setLastValueStoreTableName("logs");//记录上次采集的增量字段值的表，可以不指定，采用默认表名increament_tab
        importBuilder.setLastValueType(ImportIncreamentConfig.NUMBER_TYPE);//如果没有指定增量查询字段名称，则需要指定字段类型：ImportIncreamentConfig.NUMBER_TYPE 数字类型
        // 或者ImportIncreamentConfig.TIMESTAMP_TYPE 日期类型
        //增量配置结束

        //映射和转换配置开始
//		/**
//		 * db-es mapping 表字段名称到es 文档字段的映射：比如document_id -> docId
//		 * 可以配置mapping，也可以不配置，默认基于java 驼峰规则进行db field-es field的映射和转换
//		 */
//		importBuilder.addFieldMapping("document_id","docId")
//				.addFieldMapping("docwtime","docwTime")
//				.addIgnoreFieldMapping("channel_id");//添加忽略字段
//
//
//		/**
//		 * 为每条记录添加额外的字段和值
//		 * 可以为基本数据类型，也可以是复杂的对象
//		 */
//		importBuilder.addFieldValue("testF1","f1value");
//		importBuilder.addFieldValue("testInt",0);
//		importBuilder.addFieldValue("testDate",new Date());
//		importBuilder.addFieldValue("testFormateDate","yyyy-MM-dd HH",new Date());
//		TestObject testObject = new TestObject();
//		testObject.setId("testid");
//		testObject.setName("jackson");
//		importBuilder.addFieldValue("testObject",testObject);
//
//		/**
//		 * 重新设置es数据结构
//		 */
//		importBuilder.setDataRefactor(new DataRefactor() {
//			public void refactor(Context context) throws Exception  {
//				CustomObject customObject = new CustomObject();
//				customObject.setAuthor((String)context.getValue("author"));
//				customObject.setTitle((String)context.getValue("title"));
//				customObject.setSubtitle((String)context.getValue("subtitle"));
//				customObject.setIds(new int[]{1,2,3});
//				context.addFieldValue("docInfo",customObject);//如果还需要构建更多的内部对象，可以继续构建
//
//				//上述三个属性已经放置到docInfo中，如果无需再放置到索引文档中，可以忽略掉这些属性
//				context.addIgnoreFieldMapping("author");
//				context.addIgnoreFieldMapping("title");
//				context.addIgnoreFieldMapping("subtitle");
//			}
//		});
        //映射和转换配置结束

        /**
         * 一次、作业创建一个内置的线程池，实现多线程并行数据导入elasticsearch功能，作业完毕后关闭线程池
         */
        importBuilder.setParallel(true);//设置为多线程并行批量导入,false串行
        importBuilder.setQueue(queueSize);//设置批量导入线程池等待队列长度
        importBuilder.setThreadCount(workThreads);//设置批量导入线程池工作线程数量
        importBuilder.setContinueOnError(true);//任务出现异常，是否继续执行作业：true（默认值）继续执行 false 中断作业执行
        importBuilder.setAsyn(false);//true 异步方式执行，不等待所有导入作业任务结束，方法快速返回；false（默认值） 同步方式执行，等待所有导入作业任务结束，所有作业结束后方法才返回
        importBuilder.setEsIdField("id");//设置文档主键，不设置，则自动产生文档id
//		importBuilder.setDebugResponse(false);//设置是否将每次处理的reponse打印到日志文件中，默认false，不打印响应报文将大大提升性能，只有在调试需要的时候才打开，log日志级别同时要设置为INFO
//		importBuilder.setDiscardBulkResponse(true);//设置是否需要批量处理的响应报文，不需要设置为false，true为需要，默认true，如果不需要响应报文将大大提升处理速度

        importBuilder.setDebugResponse(false);//设置是否将每次处理的reponse打印到日志文件中，默认false
        importBuilder.setDiscardBulkResponse(true);//设置是否需要批量处理的响应报文，不需要设置为false，true为需要，默认false
        /**
         importBuilder.setEsIdGenerator(new EsIdGenerator() {
         //如果指定EsIdGenerator，则根据下面的方法生成文档id，
         // 否则根据setEsIdField方法设置的字段值作为文档id，
         // 如果默认没有配置EsIdField和如果指定EsIdGenerator，则由es自动生成文档id

         @Override public Object genId(Context context) throws Exception {
         return SimpleStringUtil.getUUID();//返回null，则由es自动生成文档id
         }
         });
         */
        /**
         * 执行数据库表数据导入es操作
         */
        DataStream dataStream = importBuilder.builder();
        dataStream.execute();//执行导入操作

        System.out.println();


    }

    /**
     * elasticsearch地址和数据库地址都从外部配置文件application.properties中获取，加载数据源配置和es配置
     */
    public void scheduleFullImportData(boolean dropIndice) {
        DB2ESImportBuilder importBuilder = DB2ESImportBuilder.newInstance();
        dropIndice = CommonLauncher.getBooleanAttribute("dropIndice", true);//同时指定了默认值
        int batchSize = CommonLauncher.getIntProperty("batchSize", 10);//同时指定了默认值
        int queueSize = CommonLauncher.getIntProperty("queueSize", 50);//同时指定了默认值
        int workThreads = CommonLauncher.getIntProperty("workThreads", 10);//同时指定了默认值
        //增量定时任务不要删表，但是可以通过删表来做初始化操作
        if (dropIndice) {
            try {
                //清除测试表,导入的时候回重建表，测试的时候加上为了看测试效果，实际线上环境不要删表
                String repsonse = ElasticSearchHelper.getRestClientUtil().dropIndice("dbdemo");
                System.out.println(repsonse);
            } catch (Exception e) {
            }
        }
        //设置数据bulk导入任务结果处理回调函数，对每次bulk任务的结果进行成功和失败反馈，然后针对失败的bulk任务通过error方法进行相应处理
        importBuilder.setExportResultHandler(new ExportResultHandler<String, String>() {
            @Override
            public void success(TaskCommand<String, String> taskCommand, String result) {
                String datas = taskCommand.getDatas();//执行的批量数据
//				System.out.println(result);//打印成功结果
            }

            @Override
            public void error(TaskCommand<String, String> taskCommand, String result) {
                //任务执行完毕，但是结果中包含错误信息
                //具体怎么处理失败数据可以自行决定,下面的示例显示重新导入失败数据的逻辑：
                // 从result中分析出导入失败的记录，然后重新构建data，设置到taskCommand中，重新导入，
                // 支持的导入次数由getMaxRetry方法返回的数字决定
                // String failDatas = ...;
                //taskCommand.setDatas(failDatas);
                //taskCommand.execute();
                String datas = taskCommand.getDatas();//执行的批量数据
//				System.out.println(result);//打印成功结果
            }

            @Override
            public void exception(TaskCommand<String, String> taskCommand, Exception exception) {
                //任务执行抛出异常，失败处理方法
            }

            /**
             * 如果对于执行有错误的任务，可以进行修正后重新执行，通过本方法
             * 返回允许的最大重试次数
             * @return
             */
            @Override
            public int getMaxRetry() {
                return -1;
            }
        });

        //指定导入数据的sql语句，必填项，可以设置自己的提取逻辑，
        // 设置增量变量log_id，增量变量名称#[log_id]可以多次出现在sql语句的不同位置中，例如：
        // select * from td_sm_log where log_id > #[log_id] and parent_id = #[log_id]
        // log_id和数据库对应的字段一致,就不需要设置setNumberLastValueColumn和setNumberLastValueColumn信息，
        // 但是需要设置setLastValueType告诉工具增量字段的类型

        importBuilder.setSql("select * from td_sm_log ");

//		importBuilder.setSql("select * from td_sm_log ");
        /**
         * es相关配置
         */
        importBuilder
                .setIndex("dbdemo") //必填项
                .setIndexType("dbdemo") //必填项
//				.setRefreshOption("refresh")//可选项，null表示不实时刷新，importBuilder.setRefreshOption("refresh");表示实时刷新
                .setUseJavaName(false) //可选项,将数据库字段名称转换为java驼峰规范的名称，true转换，false不转换，默认false，例如:doc_id -> docId
                .setUseLowcase(true)  //可选项，true 列名称转小写，false列名称不转换小写，默认false，只要在UseJavaName为false的情况下，配置才起作用
                .setPrintTaskLog(true) //可选项，true 打印任务执行日志（耗时，处理记录数） false 不打印，默认值false
                .setBatchSize(batchSize);  //可选项,批量导入es的记录数，默认为-1，逐条处理，> 0时批量处理

        //定时任务配置，
        importBuilder.setFixedRate(false)//参考jdk timer task文档对fixedRate的说明
//					 .setScheduleDate(date) //指定任务开始执行时间：日期
                .setDeyLay(1000L) // 任务延迟执行deylay毫秒后执行
                .setPeriod(10000L); //每隔period毫秒执行，如果不设置，只执行一次
        //定时任务配置结束
//
//		//设置任务执行拦截器，可以添加多个
//		importBuilder.addCallInterceptor(new CallInterceptor() {
//			@Override
//			public void preCall(TaskContext taskContext) {
//				System.out.println("preCall");
//			}
//
//			@Override
//			public void afterCall(TaskContext taskContext) {
//				System.out.println("afterCall");
//			}
//
//			@Override
//			public void throwException(TaskContext taskContext, Exception e) {
//				System.out.println("throwException");
//			}
//		}).addCallInterceptor(new CallInterceptor() {
//			@Override
//			public void preCall(TaskContext taskContext) {
//				System.out.println("preCall 1");
//			}
//
//			@Override
//			public void afterCall(TaskContext taskContext) {
//				System.out.println("afterCall 1");
//			}
//
//			@Override
//			public void throwException(TaskContext taskContext, Exception e) {
//				System.out.println("throwException 1");
//			}
//		});
//		//设置任务执行拦截器结束，可以添加多个
        //增量配置开始
//		importBuilder.setNumberLastValueColumn("log_id");//手动指定数字增量查询字段，默认采用上面设置的sql语句中的增量变量名称作为增量查询字段的名称，指定以后就用指定的字段
//		importBuilder.setDateLastValueColumn("log_id");//手动指定日期增量查询字段，默认采用上面设置的sql语句中的增量变量名称作为增量查询字段的名称，指定以后就用指定的字段
        importBuilder.setFromFirst(true);//任务重启时，重新开始采集数据，true 重新开始，false不重新开始，适合于每次全量导入数据的情况，如果是全量导入，可以先删除原来的索引数据
        importBuilder.setLastValueStorePath("logtable_import");//记录上次采集的增量字段值的文件路径，作为下次增量（或者重启后）采集数据的起点，不同的任务这个路径要不一样
//		importBuilder.setLastValueStoreTableName("logs");//记录上次采集的增量字段值的表，可以不指定，采用默认表名increament_tab
        importBuilder.setLastValueType(ImportIncreamentConfig.NUMBER_TYPE);//如果没有指定增量查询字段名称，则需要指定字段类型：ImportIncreamentConfig.NUMBER_TYPE 数字类型
        // 或者ImportIncreamentConfig.TIMESTAMP_TYPE 日期类型
        //增量配置结束

        //映射和转换配置开始
//		/**
//		 * db-es mapping 表字段名称到es 文档字段的映射：比如document_id -> docId
//		 * 可以配置mapping，也可以不配置，默认基于java 驼峰规则进行db field-es field的映射和转换
//		 */
//		importBuilder.addFieldMapping("document_id","docId")
//				.addFieldMapping("docwtime","docwTime")
//				.addIgnoreFieldMapping("channel_id");//添加忽略字段
//
//
//		/**
//		 * 为每条记录添加额外的字段和值
//		 * 可以为基本数据类型，也可以是复杂的对象
//		 */
//		importBuilder.addFieldValue("testF1","f1value");
//		importBuilder.addFieldValue("testInt",0);
//		importBuilder.addFieldValue("testDate",new Date());
//		importBuilder.addFieldValue("testFormateDate","yyyy-MM-dd HH",new Date());
//		TestObject testObject = new TestObject();
//		testObject.setId("testid");
//		testObject.setName("jackson");
//		importBuilder.addFieldValue("testObject",testObject);
//
//		/**
//		 * 重新设置es数据结构
//		 */
//		importBuilder.setDataRefactor(new DataRefactor() {
//			public void refactor(Context context) throws Exception  {
//				CustomObject customObject = new CustomObject();
//				customObject.setAuthor((String)context.getValue("author"));
//				customObject.setTitle((String)context.getValue("title"));
//				customObject.setSubtitle((String)context.getValue("subtitle"));
//				customObject.setIds(new int[]{1,2,3});
//				context.addFieldValue("docInfo",customObject);//如果还需要构建更多的内部对象，可以继续构建
//
//				//上述三个属性已经放置到docInfo中，如果无需再放置到索引文档中，可以忽略掉这些属性
//				context.addIgnoreFieldMapping("author");
//				context.addIgnoreFieldMapping("title");
//				context.addIgnoreFieldMapping("subtitle");
//			}
//		});
        //映射和转换配置结束

        /**
         * 一次、作业创建一个内置的线程池，实现多线程并行数据导入elasticsearch功能，作业完毕后关闭线程池
         */
        importBuilder.setParallel(true);//设置为多线程并行批量导入,false串行
        importBuilder.setQueue(queueSize);//设置批量导入线程池等待队列长度
        importBuilder.setThreadCount(workThreads);//设置批量导入线程池工作线程数量
        importBuilder.setContinueOnError(true);//任务出现异常，是否继续执行作业：true（默认值）继续执行 false 中断作业执行
        importBuilder.setAsyn(false);//true 异步方式执行，不等待所有导入作业任务结束，方法快速返回；false（默认值） 同步方式执行，等待所有导入作业任务结束，所有作业结束后方法才返回
        importBuilder.setEsIdField("log_id");//设置文档主键，不设置，则自动产生文档id
//		importBuilder.setDebugResponse(false);//设置是否将每次处理的reponse打印到日志文件中，默认false，不打印响应报文将大大提升性能，只有在调试需要的时候才打开，log日志级别同时要设置为INFO
//		importBuilder.setDiscardBulkResponse(true);//设置是否需要批量处理的响应报文，不需要设置为false，true为需要，默认true，如果不需要响应报文将大大提升处理速度

        importBuilder.setDebugResponse(false);//设置是否将每次处理的reponse打印到日志文件中，默认false
        importBuilder.setDiscardBulkResponse(true);//设置是否需要批量处理的响应报文，不需要设置为false，true为需要，默认false
        /**
         importBuilder.setEsIdGenerator(new EsIdGenerator() {
         //如果指定EsIdGenerator，则根据下面的方法生成文档id，
         // 否则根据setEsIdField方法设置的字段值作为文档id，
         // 如果默认没有配置EsIdField和如果指定EsIdGenerator，则由es自动生成文档id

         @Override public Object genId(Context context) throws Exception {
         return SimpleStringUtil.getUUID();//返回null，则由es自动生成文档id
         }
         });
         */
        /**
         * 执行数据库表数据导入es操作
         */
        DataStream dataStream = importBuilder.builder();
        dataStream.execute();//执行导入操作

        System.out.println();


    }
}
